package com.zlx.rulemk.marketing.utils;


import com.typesafe.config.ConfigFactory;
import com.zlx.rulemk.marketing.beans.DynamicKeyRuleStateBean;
import com.zlx.rulemk.marketing.beans.EventBean;
import com.zlx.rulemk.marketing.beans.RuleMatchResult;
import com.zlx.rulemk.marketing.beans.RuleStateBean;
import com.zlx.rulemk.marketing.constant.ConfigNames;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.OutputTag;

import java.util.List;

/**
 * @Author: zhanglingxing
 * @CreateTime: 2022-07-07
 * @Description: Flink状态描述器
 * @Version: 1.0
 */
public class FlinkStateDescUtils {

    /**
     * 近期行为事件存储状态描述器
     */
    public static ListStateDescriptor<EventBean> getEventBeansDesc(){
        ListStateDescriptor<EventBean> eventBeansDesc = new ListStateDescriptor<>("event_beans", EventBean.class);
        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(2)).build();
        eventBeansDesc.enableTimeToLive(ttlConfig);


        return eventBeansDesc;
    }


    public static ListStateDescriptor<Tuple2<String, Long>> ruleTimerStateDesc
            =  new ListStateDescriptor<Tuple2<String, Long>>("rule_timer", TypeInformation.of(new TypeHint<Tuple2<String, Long>>() {}));


    // <触发事件ID，list<多条规则分组对象>>
    public static MapStateDescriptor<String,  List<DynamicKeyRuleStateBean> > ruleStateDesc1
            = new MapStateDescriptor<>("rule_broadcast_state_1"
            ,TypeInformation.of(String.class)
            ,TypeInformation.of(new TypeHint<List<DynamicKeyRuleStateBean>>() {}));


    // <分组keyby字段名,null>
    public static ListStateDescriptor<RuleStateBean> ruleStateDesc2 = new ListStateDescriptor<>("rule_broadcast_state_2", RuleStateBean.class);


    //将数据进行拆分 原来使用split方式，再select，现在使用侧流输出 有定时条件的的流
    public static OutputTag<RuleMatchResult> hasOnTimerStream = new OutputTag<RuleMatchResult>("has-ontimer-output") {};

    //没有定时条件的侧输出流
    public static OutputTag<RuleMatchResult> notOnTimerStream = new OutputTag<RuleMatchResult>("not-ontimer-output") {};
}
